package com.ddxz.data.word;

import com.ddxz.data.word.join.DdxzJoin;
import com.ddxz.data.word.map.DdxzSplitterMap;
import com.ddxz.data.word.struct.FirstData;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.types.Row;

import javax.sql.DataSource;

/**
 * class description
 *
 * @author 朱宗刚
 * @version 1.0.0
 * @date 2020/9/1
 */

public class WordApplication {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStream<Tuple2<String, Integer>> ds = env.readTextFile("D:\\move\\apache-maven-3.6.3\\conf\\settings.xml")
//                .flatMap(new DdxzSplitterMap())
//                .keyBy(value -> value.f0)
////                .timeWindow(Time.seconds(5))
//                .sum(1);

        JDBCInputFormat inputFormat1 = JDBCInputFormat.buildJDBCInputFormat()
                .setDBUrl("jdbc:mysql://192.168.12.140:3306/demo?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setUsername("root")
                .setPassword("root")
                .setQuery("SELECT c_id, c_user, c_status, c_create_time, c_real_name, c_schema from t_meta_object")
                .setRowTypeInfo(new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),
                        TypeInformation.of(Integer.class), TypeInformation.of(Long.class),
                        TypeInformation.of(String.class), TypeInformation.of(String.class)))
                .finish();
        JDBCInputFormat inputFormat2 = JDBCInputFormat.buildJDBCInputFormat()
                .setDBUrl("jdbc:mysql://192.168.12.140:3306/dmp?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setUsername("root")
                .setPassword("root")
                .setQuery("select c_id, c_rid, c_real_name, c_display_name from t_meta_column")
                .setRowTypeInfo(new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class),
                        TypeInformation.of(String.class), TypeInformation.of(String.class)))
                .finish();
//        ds.print();

        inputFormat1.openInputFormat();
        inputFormat2.openInputFormat();

        DataStreamSource<Row> dss1 = env.createInput(inputFormat1);
        DataStreamSource<Row> dss2 = env.createInput(inputFormat2);
        dss1.print();
        dss2.print();
        dss1.join(dss2).where(i1 -> i1.getField(1)).equalTo(i2-> i2.getField(2)).window(TumblingProcessingTimeWindows.of(Time.seconds(60)));
//                .apply(new DdxzJoin() );

        inputFormat1.closeInputFormat();
        inputFormat2.closeInputFormat();

        env.execute("ddxz word count");
    }
}
